package com.atguigu.day03;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

// 自定义并行数据源
// +--------------+    +---------------+
// |      run     |    |      run      |
// +--------------+    +---------------+
//       \                      /
//        \                    /
//         v                  v
//        +-------------------+
//        |       print       |
//        +-------------------+
public class Example11 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env
                .addSource(new RichParallelSourceFunction<Integer>() {
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        int subTaskIndex = getRuntimeContext().getIndexOfThisSubtask();
                        System.out.println("索引：" + subTaskIndex + " 的生命周期开始了");
                    }

                    // 有两个并行子任务执行run函数
                    @Override
                    public void run(SourceContext<Integer> ctx) throws Exception {
                        int subTaskIndex = getRuntimeContext().getIndexOfThisSubtask();
                        for (int i = 1; i < 9; i++) {
                            // 两个并行数据源都要执行for循环
                            // 但两个并行数据源的子任务索引不一样，一个是0,一个是1
                            // 所以索引为0的并行数据源发送偶数
                            // 索引为1的并行数据源发送奇数
                            if (i % 2 == subTaskIndex) {
                                ctx.collect(i);
                            }
                        }
                    }

                    @Override
                    public void cancel() {

                    }

                    @Override
                    public void close() throws Exception {
                        super.close();
                        int subTaskIndex = getRuntimeContext().getIndexOfThisSubtask();
                        System.out.println("索引：" + subTaskIndex + " 的生命周期结束了");
                    }
                })
                .setParallelism(2)
                .print()
                .setParallelism(2);

        env.execute();
    }
}
